package com.dahua.messagesyncsh.rabbitmq;


import com.dahua.messagesyncsh.config.redis.RedisUtil;
import com.dahua.messagesyncsh.rabbitmq.enums.Rabbitmq;
import com.raizumi.common.utils.tool.ContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.ErrorHandler;

import javax.annotation.Resource;
import java.util.Optional;
import java.util.Properties;

import static com.dahua.messagesyncsh.common.Constants.RABBITMQ_CONSUMER_CONTAINER_NAME;

@Slf4j
@Component
public class RabbitmqUtil {

    @Resource
    @Qualifier("RabbitTemplate")
    private RabbitTemplate rabbitTemplate;

    @Resource
    private RabbitAdmin rabbitAdmin;

    @Resource
    private ContextUtil contextUtil;

    @Resource
    private RedisUtil redisUtil;

    public void send(String queueName, String message) {
        rabbitTemplate.convertAndSend(queueName, message);
    }

    /*  创建队列绑定主机和主题*/
    public void create(String queueName, String exchangeName, String routingKey) {
        QueueInformation queueInfo = rabbitAdmin.getQueueInfo(queueName);
        Queue queue = QueueBuilder.durable(queueName).build();
        rabbitAdmin.declareQueue(queue);
        Exchange exchange = ExchangeBuilder.topicExchange(exchangeName).durable(true).build();
        Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey).noargs();
        rabbitAdmin.declareBinding(binding);
    }

    public boolean registerConsumer(Properties properties,
                                    MessageListener messageListener,
                                    ErrorHandler errorHandler,
                                    PlatformTransactionManager transactionManager) {
        SimpleMessageListenerContainer container = new RabbitmqConfig().messageListenerContainer(properties, messageListener, errorHandler, transactionManager);
        if (container == null) {
            log.error("【rabbitmq】：consumer bean为空");
            return false;
        } else {
            String property = properties.getProperty(Rabbitmq.QUEUE.getName());
            if (property == null){
                log.error("【rabbitmq】：queue未配置");
                return false;
            }
            return contextUtil.register(property, SimpleMessageListenerContainer.class ,container);
        }
    }

    public boolean unregisterConsumer(String containerName){
        return contextUtil.unregister(containerName);
    }

    public void startConsumer(String containerName) {
        Optional<SimpleMessageListenerContainer> bean = contextUtil.getBean(containerName, SimpleMessageListenerContainer.class);
        if (!bean.isPresent()) {
            return;
        }
        SimpleMessageListenerContainer container = bean.get();
        container.start();
        redisUtil.setHash(RABBITMQ_CONSUMER_CONTAINER_NAME, containerName, true);
    }

    public void stopConsumer(String containerName) {
        Optional<SimpleMessageListenerContainer> bean = contextUtil.getBean(containerName, SimpleMessageListenerContainer.class);
        if (!bean.isPresent()) {
            return;
        }
        SimpleMessageListenerContainer container = bean.get();
        container.stop();
        redisUtil.setHash(RABBITMQ_CONSUMER_CONTAINER_NAME, containerName, false);
    }

}
